package com.ydl.learning.flink.demo.time

import com.ydl.learning.flink.demo.Utils
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * Event Time: 事件产生的时间,它通常由事件中的时间戳描述
 * 数据从终端产生,或者从系统中产生的过程中生成的时间为事件生成时间
 * 默认情况下使用的是Process Time(事件被处理时当前系统的时间)时间语义
 *
 * 使用自定义的水位线类MyWaterMark
 *
 * @author ydl
 * @since 2020/11/23
 */
object EventTimeDemo extends App with Utils {
  //配制处理数据的时候使用event time
  streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  //周期性地生成Watermark,默认是100ms,每隔N毫秒自动向流里注入一个Watermark
  streamEnv.getConfig.setAutoWatermarkInterval(1000)
  //添加自定义的 TimestampsAndWatermarks
  //一种更简单的方法是定义一个匿名的BoundedOutOfOrdernessTimestampExtractor类，例如
  //延迟2s
  //new BoundedOutOfOrdernessTimestampExtractor[StationLog](Time.seconds(2)) {
  //  override def extractTimestamp(t: StationLog): Long = {
  //    t.callTime
  //  }
  //}
  consumer.assignTimestampsAndWatermarks(new MyWaterMark)
  val stream = streamEnv.addSource(consumer)
    .map(str => {
      val strings = str.split(" ")
      (strings(2), Integer.valueOf(strings(1)))
    })
    .keyBy(_._1)
    .timeWindow(Time.seconds(5))
    //.reduce((a, b) => (a._1, a._2 + b._2))
  //stream.print()
  streamEnv.execute("EventTimeDemo")

}

